package com.whut.monitor.dao.impl;


import com.whut.monitor.dao.IHBaseDao;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

/**
 * @author Sandeepin
 *         HBase操作底层接口
 */
@Component
public class HBaseDaoImpl implements IHBaseDao {
    private static final Logger logger = Logger.getLogger(HBaseDaoImpl.class);

    @Resource(name = "hbaseTemplate")
    private HbaseTemplate hbaseTemplate;

    /**
     * 添加单条数据
     *
     * @param tableName 表名
     * @param rowKey    行健
     * @param family    列族
     * @param qualifier 列
     * @param value     值
     * @return 是否添加成功
     */
    @Override
    public Boolean add(String tableName, final String rowKey, final String family, final String qualifier, final String value) {
        return hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {
            @Override
            public Boolean doInTable(HTableInterface table) throws Throwable {
                boolean flag = false;
                try {
                    byte[] rowkey = rowKey.getBytes();
                    Put put = new Put(rowkey);
                    put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
                    table.put(put);
                    flag = true;
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return flag;
            }
        });
    }

    /**
     * 查询单条数据
     *
     * @param tableName 表名
     * @param rowKey    行健
     * @param family    列族
     * @param qualifier 列
     * @return 返回查询到的Value
     */
    @Override
    public String find(String tableName, String rowKey, final String family, final String qualifier) {
        String value = hbaseTemplate.get(tableName, rowKey, new RowMapper<String>() {
            @Override
            public String mapRow(Result result, int rowNum) throws Exception {
                return Bytes.toString(result.getValue(family.getBytes(), qualifier.getBytes()));
            }
        });
        return value;
    }

    /**
     * 批量增加 Qua 和 Value 到 Hbase
     *
     * @param tableName   表名
     * @param rowKey      行健
     * @param family      列族
     * @param quaAndValue 列与值的Map
     * @return 是否添加成功
     */
    @Override
    public Boolean adds(String tableName, final String rowKey, final String family, final Map<String, String> quaAndValue) {
        boolean b = hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {
                    @Override
                    public Boolean doInTable(HTableInterface table) throws Throwable {
                        boolean flag = false;
                        try {
                            byte[] rowkey = rowKey.getBytes();
                            Put put = new Put(rowkey);
                            // 批量添加列与值
                            for (String aQua : quaAndValue.keySet()) {
                                put.add(Bytes.toBytes(family), Bytes.toBytes(aQua), Bytes.toBytes(quaAndValue.get(aQua)));
                            }
                            table.put(put);
                            flag = true;
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        return flag;
                    }
                }
        );
        return b;
    }

    /**
     * 批量增加 RowKey 和 Value 到 Hbase
     *
     * @param tableName      表名
     * @param rowkeyAndValue 行健与值的Map
     * @param family         列族
     * @param qualifier      列
     * @return 是否添加成功
     */
    @Override
    public Boolean adds(String tableName, final Map<String, String> rowkeyAndValue, final String family, final String qualifier) {
        boolean b = hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {
                    @Override
                    public Boolean doInTable(HTableInterface table) throws Throwable {
                        boolean flag = false;
                        try {
                            // table.setAutoFlushTo(true);
                            List<Put> lists = new ArrayList<>();
                            // 批量写RowKey和Value
                            for (String aRowkey : rowkeyAndValue.keySet()) {
                                Put put = new Put(aRowkey.getBytes());
                                put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(rowkeyAndValue.get(aRowkey)));
                                lists.add(put);
                            }
                            table.put(lists);
                            // 暂未知道关闭的影响
                            // table.close();
                            flag = true;
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        return flag;
                    }
                }
        );
        return b;
    }

    /**
     * 由 RowKey开始和结束位置 批量批量查询Value(列和列族定死)
     *
     * @param tableName            表名
     * @param startRow             行健-开始
     * @param stopRow              行健-结束
     * @param outputRowkeyAndValue RowKey和对应存储内容
     * @return 是否查找成功
     */
    @Override
    public Boolean finds(String tableName, String startRow, String stopRow, final Map<String, String> outputRowkeyAndValue) {
        long startTime = System.currentTimeMillis();
        Scan scan = new Scan();
        if (startRow == null) {
            startRow = "";
        }
        if (stopRow == null) {
            stopRow = "";
        }
        // RowKey扫描范围
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(stopRow));
        hbaseTemplate.find(tableName, scan, new RowMapper<Map<String, Object>>() {
            @Override
            public Map<String, Object> mapRow(Result result, int rowNum) throws Exception {
                List<Cell> ceList = result.listCells();
                Map<String, Object> map = new HashMap<>();
                Map<String, Map<String, Object>> returnMap = new HashMap<String, Map<String, Object>>();
                String row = "";
                if (ceList != null && ceList.size() > 0) {
                    for (Cell cell : ceList) {
                        // 取RowKey
                        row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                        // 取值
                        String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                        // 列族目前定死一个,不用滤除
                        //String family = Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength());
                        // 列目前定死一个,不用滤除
                        //String quali = Bytes.toString( cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());
                        // 输出的ID号以时间和传感器唯一ID构成,用,号间隔
                        // map中key以rowkey构成, 值是数据集合
                        map.put(row, value);
                    }
                }

                for (String key : map.keySet()) {
                    outputRowkeyAndValue.put(key, (String) map.get(key));
                }
                return map;
            }
        });
        long endTime = System.currentTimeMillis();
        System.out.println("运行时间(纯粹查Hbase):" + (endTime - startTime) + "ms 约" + (endTime - startTime) / 1000 + "s");
        return true;
    }

}
